package com.zs.juc;

import java.util.*;
import java.util.concurrent.*;

/**
 * @auther: madisonzhuang
 * @date: 2020/3/10 11:23
 * @description:
 */
public class CompletionServiceTest {

    public static void main(String[] args) {
        test();
//        test1();
//        test2();
//        test3();
    }

    /**
     * 通过ExecutorService我们可以提交一个个的task，并且返回Future，然后通过调用Future.get方法来返回任务的执行结果。
     * <p>
     * 这种方式虽然有效，但是需要保存每个返回的Future值，还是比较麻烦的，幸好ExecutorService提供了一个invokeAll的方法，来保存所有的Future值
     * 除了invokeAll方法外，我们今天要介绍一个CompletionService接口。
     *
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static void test() {
        ExecutorService executor = Executors.newFixedThreadPool(10);

        Callable<String> callableTask = () -> {
            TimeUnit.MILLISECONDS.sleep(3000);
            return "Task's execution";
        };

        List<Callable<String>> callableTasks = new ArrayList<>();
        callableTasks.add(callableTask);
        callableTasks.add(callableTask);
        callableTasks.add(callableTask);

        try {
            List<Future<String>> futures = executor.invokeAll(callableTasks);
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
    }

    public static void test1() {
        Long start = System.currentTimeMillis();
        //开启3个线程
        ExecutorService exs = Executors.newFixedThreadPool(8);
        try {
            int taskCount = 5;
            // 结果集
            List<Future<Integer>> futureList = new ArrayList<>();

            // 1.定义CompletionService
            CompletionService<Integer> completionService = new ExecutorCompletionService<>(exs);

            // 2.添加任务
            for (int i = 0; i < taskCount; i++) {
                Future<Integer> future = completionService.submit(new Task(i + 1));
                futureList.add(future);
            }

            // 3.获取结果
            List<Integer> list = new ArrayList<>();
            for (int i = 0; i < taskCount; i++) {
                Integer result = completionService.take().get();
                System.out.println("任务i=" + result + "完成!" + System.currentTimeMillis());
                list.add(result);
            }

            System.out.println("list=" + list);

            List<Integer> l = new ArrayList<>();
            for (Future<Integer> future : futureList) {
                Integer integer = future.get();
                l.add(integer);
            }
            System.out.println("l=" + l);

            System.out.println(System.currentTimeMillis() - start);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭线程池
            exs.shutdown();
        }
    }

    public static void test2() {
        int taskCount = 5;
        Random random = new Random();
        ExecutorService pool = Executors.newFixedThreadPool(8);

        CompletionService<String> service = new ExecutorCompletionService<>(pool);
        try {
            for (int i = 0; i < taskCount; i++) {
                int finalI = i;
                service.submit(() -> {
                    Thread.sleep(random.nextInt(1000));
                    System.out.println(Thread.currentThread().getName() + "|完成任务" + finalI);
                    return "data" + finalI;
                });
            }

            for (int j = 0; j < taskCount; j++) {
                Future<String> take = service.take(); //这一行没有完成的任务就阻塞
//                Future<String> take = service.poll(); // 不等待，有结果就返回一个 Future 对象，否则返回 null。
                if (Objects.nonNull(take)) {
                    String result = take.get(); // 这一行在这里不会阻塞，引入放入队列中的都是已经完成的任务
                    System.out.println("获取到结果：" + result);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            pool.shutdown();
        }
    }

    public static void test3() {
        int taskCount = 5;
        ExecutorService pool = Executors.newFixedThreadPool(5);
        List<Future<String>> resultFuture = new ArrayList<>();

        try {
            for (int i = 0; i < taskCount; i++) {
                int tmp = i;
                Future<String> future = pool.submit(() -> {
                    Thread.sleep(1000 + 10 * tmp);
                    System.out.println(Thread.currentThread().getName() + "|完成任务" + tmp);
                    return "data" + tmp;
                });
                resultFuture.add(future);
            }

            for (Future<String> future : resultFuture) {
                String result = future.get();
                System.out.println("执行结果" + result);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            pool.shutdown();
        }
    }

    static class Task implements Callable<Integer> {
        Integer i;

        public Task(Integer i) {
            super();
            this.i = i;
        }

        @Override
        public Integer call() throws Exception {
            if (i == 5) {
                Thread.sleep(5000);
            } else {
                Thread.sleep(10000);
            }
            System.out.println("线程：" + Thread.currentThread().getName() + "任务i=" + i + ",执行完成！");
            return i;
        }

    }
}
